package com.wandian.executor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.wandian.util.Log;

/**
 * @author long
 * @Create 2012-11-13 下午3:06:57
 * @QQ 108232706
 * @version 1.0
 * @description Executor
 */
public class Executor {
	private final AbstractActionQueue defaultQueue;
	private final ThreadPoolExecutor pool;
	private final DelayCheckThread delayCheckThread;

	public Executor(int corePoolSize, int maxPoolSize, int keepAliveTime, int cacheSize, String prefix)
	{
		TimeUnit unit = TimeUnit.MINUTES;
		LinkedBlockingQueue workQueue = new LinkedBlockingQueue<Action>();
		RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
		if (prefix == null) {
			prefix = "";
		}
		ThreadFactory threadFactory = new Threads(prefix);
		this.pool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
		this.defaultQueue = new AbstractActionQueue(this);
		this.delayCheckThread = new DelayCheckThread(prefix);
		this.delayCheckThread.start();
	}

	public AbstractActionQueue getDefaultQueue() {
		return this.defaultQueue;
	}

	public void enDefaultQueue(Action action) {
		this.defaultQueue.enqueue(action);
	}

	public void execute(Action action) {
		this.pool.execute(action);
	}

	public void run(Runnable action) {
		Log.debug("执行动作");
		this.pool.execute(action);
	}

	public void enDelayQueue(DelayAction delayAction) {
		this.delayCheckThread.addAction(delayAction);
	}

	public void stop() {
		this.delayCheckThread.stopping();
		if (!this.pool.isShutdown())
			this.pool.shutdown();
	}

	class DelayCheckThread extends Thread
	{
		private static final int FRAME_PER_SECOND = 120;
		private final Object lock = new Object();
		private final List<DelayAction> queue;
		private final List<DelayAction> execQueue;
		private boolean isRunning;

		public DelayCheckThread(String prefix)
		{
			super("DelayCheckThread-" + prefix);
			this.queue = new ArrayList<DelayAction>();
			this.execQueue = new ArrayList<DelayAction>();
			this.isRunning = true;
			setPriority(10);
		}

		public boolean isRunning() {
			return this.isRunning;
		}

		public void stopping() {
			if (this.isRunning)
				this.isRunning = false;
		}

		@Override
		public void run()
		{
			long balance = 0L;
			while (this.isRunning)
				try {
					int execute = 0;
					poll();
					if (this.execQueue.size() != 0)
					{
						long start = System.currentTimeMillis();
						execute = execActions();
						this.execQueue.clear();
						long end = System.currentTimeMillis();
						long interval = end - start;
						balance += 120L - interval;
						if (interval > 120L) {
							Log.warn("DelayCheckThread is spent too much time: " + interval + "ms, execute = " + execute);
						}
						if (balance > 0L) {
							Thread.sleep((int) balance);
							balance = 0L;
						}
						else if (balance < -1000L) {
							balance += 1000L;
						}
					}
				} catch (Exception e) {
					Log.error("DelayCheckThread error. ", e);
				}
		}

		public int execActions()
		{
			int executeCount = 0;
			for (DelayAction delayAction : this.execQueue) {
				try {
					long begin = System.currentTimeMillis();
					if (delayAction == null) {
						Log.error("error");
					}
					if (!delayAction.canExec(begin)) {
						addAction(delayAction);
					}
					executeCount++;
					long end = System.currentTimeMillis();
					if (end - begin > 120L)
						Log.warn(delayAction.toString() + " spent too much time. time :" + (end - begin));
				} catch (Exception e) {
					Log.error("执行action异常" + delayAction.toString(), e);
				}
			}
			return executeCount;
		}

		public void addAction(DelayAction delayAction)
		{
			synchronized (this.lock) {
				this.queue.add(delayAction);
				this.lock.notifyAll();
			}
		}

		private void poll()
				throws InterruptedException
		{
			synchronized (this.lock) {
				if (this.queue.size() == 0) {
					this.lock.wait();
				} else {
					this.execQueue.addAll(this.queue);
					this.queue.clear();
					this.lock.notifyAll();
				}
			}
		}
	}

	static class Threads implements ThreadFactory
	{
		static final AtomicInteger poolNumber = new AtomicInteger(1);
		final ThreadGroup group;
		final AtomicInteger threadNumber = new AtomicInteger(1);
		final String namePrefix;

		public Thread newThread(Runnable runnable)
		{
			Thread thread = new Thread(this.group, runnable, this.namePrefix + this.threadNumber.getAndIncrement(), 0L);
			if (thread.isDaemon())
				thread.setDaemon(false);
			if (thread.getPriority() != 5)
				thread.setPriority(5);
			return thread;
		}

		Threads(String prefix) {
			SecurityManager securitymanager = System.getSecurityManager();
			this.group = (securitymanager == null ? Thread.currentThread().getThreadGroup() : securitymanager.getThreadGroup());
			this.namePrefix = ("pool-" + poolNumber.getAndIncrement() + "-" + prefix + "-thread-");
		}
	}
}